/***************************************************************************
 * Copyright 2012-2013 TXT e-solutions SpA
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * This work was performed within the IoT_at_Work Project
 * and partially funded by the European Commission's
 * 7th Framework Programme under the contract ICT-257367.
 *
 * Authors:
 *      Salvatore Piccione (TXT e-solutions SpA)
 *
 * Contributors:
 *        Domenico Rotondi (TXT e-solutions SpA)
 **************************************************************************/
package it.txt.ens.client.subscriber.impl;

import it.txt.ens.client.core.ENSBrokerConnectionParameters;
import it.txt.ens.client.core.ENSEvent;
import it.txt.ens.client.core.factory.ENSBrokerConnectionParametersFactory;
import it.txt.ens.client.impl.BasicENSClient;
import it.txt.ens.client.subscriber.ENSEventListener;
import it.txt.ens.client.subscriber.ENSSubscriber;
import it.txt.ens.client.subscriber.ENSSubscriptionException;
import it.txt.ens.client.subscriber.factory.ENSEventFactory;
import it.txt.ens.client.subscriber.factory.ENSSubscriberFactory;
import it.txt.ens.client.util.AMQPConstants;
import it.txt.ens.core.ENSAuthzServiceConnectionParameters;
import it.txt.ens.core.ENSOperation;
import it.txt.ens.core.ENSResource;
import it.txt.ens.core.KeystoreParameters;
import it.txt.ens.core.X509CertificateRetrievalParameters;
import it.txt.ens.schema.request.factory.ENSAuthorisationRequestFactory;
import it.txt.ens.schema.response.factory.ENSAuthorisationResponseFactory;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * Provides the default implementation of {@link ENSSubscriber}. 
 * 
 * @author Salvatore Piccione (TXT e-solutions SpA - salvatore.piccione AT txtgroup.com)
 * @author Domenico Rotondi (TXT e-solutions SpA - domenico.rotondi AT txtgroup.com)
 *
 */
public class BasicENSSubscriber extends BasicENSClient implements ENSSubscriber {
    /**
     * The ID of this {code ENSSubscriber} implementation
     */
    public static final String IMPLEMENTATION_ID = "default";
    private static final Logger LOGGER = Logger.getLogger(BasicENSSubscriber.class.getName(),"logging-messages/" + 
        BasicENSSubscriber.class.getSimpleName());
//    private static final int MESSAGE_DELIVERY_TIMEOUT = 1000;
    
    private String queueName;
    private Map<String, ENSEventListener> listeners = new HashMap<String, ENSEventListener>();
    private DefaultConsumer basicConsumer;
    private boolean isSubscribed;
    private final ENSEventFactory eventFactory;
    
    private BasicENSSubscriber(String subjectID, ENSResource targetResource,
            Date sessionExpiration,
            ENSAuthzServiceConnectionParameters accessRequestBrokerConnParams,
            KeystoreParameters keystoreParams,
            X509CertificateRetrievalParameters certParams,
            File capabilityDirectory,
            ENSAuthorisationRequestFactory requestFactory,
            ENSAuthorisationResponseFactory responseFactory, 
            ENSBrokerConnectionParametersFactory connParamsFactory, 
            ENSEventFactory eventFactory) {
        super(subjectID, targetResource, ENSOperation.subscribe, sessionExpiration, accessRequestBrokerConnParams,
                keystoreParams, certParams, capabilityDirectory, requestFactory,
                responseFactory, connParamsFactory);
        if (eventFactory == null)
            throw new IllegalArgumentException("The event factory cannot be null");
        this.eventFactory = eventFactory;
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#subscribe()
     */
    @Override
    public void subscribe() throws IllegalStateException, ENSSubscriptionException {
        if (isSubscribed)
            throw new IllegalStateException ("A subscription is already active");
        
        try {
            //create the connection
            if (!connection.isOpen())
                connection = factory.newConnection();
            channel = connection.createChannel();
            
            basicConsumer = new BasicENSConsumer(channel);
//            eventConsumer = new QueueingConsumer(channel);
//            channel.basicConsume(queueName, eventConsumer);
            
            String consumerTag = channel.basicConsume(queueName, true, basicConsumer);
            if (LOGGER.isLoggable(Level.INFO)) {
                LOGGER.log(Level.INFO, "subscription.OK", new String[] {
                        consumerTag, this.targetResource.getNamespace(), this.targetResource.getPattern()});
            }
            
//            eventConsumerThread = new Thread(this);
//            eventConsumerThread.start();
            isSubscribed = true;
        } catch (IOException e) {
            ENSSubscriptionException ese = new ENSSubscriptionException("subscription.FAILURE", e,
                this.targetResource.getNamespace(), this.targetResource.getPattern());
            if (LOGGER.isLoggable(Level.SEVERE))
                LOGGER.log(Level.SEVERE, ese.getMessage(), e);
            throw ese;
        }
        
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#unsubscribe()
     */
    @Override
    public void unsubscribe() throws IllegalStateException, ENSSubscriptionException {
        if (!isSubscribed)
            throw new IllegalStateException ("No subscription active");
        this.isSubscribed = false;
//            eventConsumerThread.join();
//            channel.basicCancel(eventConsumer.getConsumerTag());
        try {
            channel.basicCancel(basicConsumer.getConsumerTag());
            channel.close();
        } catch (IOException e) {
            ENSSubscriptionException ese = new ENSSubscriptionException("unsubscription.FAILURE", e, 
                basicConsumer.getConsumerTag(), targetResource.getNamespace(), targetResource.getPattern());
            if (LOGGER.isLoggable(Level.SEVERE))
                LOGGER.log(Level.SEVERE, ese.getMessage(), e);
            closeConnection();
            throw ese;
        }
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#addEventListener(it.txt.ens.client.subscriber.ENSEventListener)
     */
    @Override
    public String registerEventListener(ENSEventListener listener) {
        if (listener == null)
            throw new IllegalArgumentException ("The listener cannot be null");
        UUID listenerID = UUID.randomUUID();
        String listenerID_STR = listenerID.toString();
        listeners.put(listenerID.toString(), listener);
        return listenerID_STR;
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#removeEventListener(java.lang.String)
     */
    @Override
    public void unregisteredEventListener(String listenerID) {
        if (listeners.remove(listenerID) == null)
            throw new IllegalArgumentException("No listener associated to ID " + listenerID);
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#removeEventListeners(java.lang.String[])
     */
    @Override
    public void unregisterEventListeners(String... listenerIDs) {
        if (listenerIDs == null)
            throw new IllegalArgumentException("The array of the listener's IDs to be removed cannot be null");
        if (listenerIDs.length == 0)
            throw new IllegalArgumentException("The array of the listener's IDs to be removed cannot be empty");
        for (int i = 0; i < listenerIDs.length; i++)
            unregisteredEventListener(listenerIDs[i]);
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#removeEventListeners(java.util.List)
     */
    @Override
    public void unregisterEventListeners(Collection<String> listenerIDs) {
        if (listenerIDs == null)
            throw new IllegalArgumentException("The collection of the listener's IDs to be removed cannot be null");
        if (listenerIDs.isEmpty())
            throw new IllegalArgumentException("The collection of the listener's IDs to be removed cannot be empty");
        Iterator<String> i = listenerIDs.iterator();
        while (i.hasNext())
            unregisteredEventListener(i.next());
    }

    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
//    @Override
//    public void run() {
//        try {
//            Delivery delivery;
//            while (isSubscribed) {
//                delivery = eventConsumer.nextDelivery(MESSAGE_DELIVERY_TIMEOUT);
//                if (delivery != null) {
//                    //notifies the listeners
//                    for (ENSEventListener listener : listeners.values())
//                        listener.onEvent(parseMessage(delivery));
//                }
//            }
//        } catch (InterruptedException e) {
//            
//        } catch (ShutdownSignalException e) {
//            
//        } catch (ConsumerCancelledException e) {
//            
//        }
//    }

//    private ENSEventToBeConsumed parseMessage (Delivery message) {
//        Envelope envelope = message.getEnvelope();
//        BasicProperties properties = message.getProperties();
//        return eventFactory.create(envelope.getRoutingKey(), properties.getHeaders(), 
//            message.getBody(), properties.getContentType(), properties.getContentEncoding(),
//            properties.getPriority(), properties.getTimestamp(), properties.getUserId(), properties.getAppId(), false);
//    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.impl.BasicENSClient#useENSBrokerConnectionParameters(it.txt.ens.client.core.ENSBrokerConnectionParameters)
     */
    @Override
    protected void useENSBrokerConnectionParameters(
            ENSBrokerConnectionParameters parameters) {
        this.queueName = parameters.getDestinationName();
    }

    /* (non-Javadoc)
     * @see it.txt.ens.client.subscriber.ENSSubscriber#isSubscribed()
     */
    @Override
    public boolean isSubscribed() {
        return isSubscribed;
    }
    /**
     * Provides the default implementation of {@link ENSSubscriberFactory}<br/>It creates
     * {@link BasicENSSubscriber} objects.
     * @author Salvatore Piccione (TXT e-solutions SpA - salvatore.piccione AT txtgroup.com)
     * @author Domenico Rotondi (TXT e-solutions SpA - domenico.rotondi AT txtgroup.com)
     *
     */
    public static class BasicENSSubscriberFactory implements ENSSubscriberFactory {

        /* (non-Javadoc)
         * @see it.txt.ens.client.subscriber.factory.ENSSubscriberFactory#create(java.lang.String, it.txt.ens.core.ENSResource, it.txt.ens.core.ENSConnectionParameters, it.txt.ens.core.ENSConnectionParameters, it.txt.ens.core.KeystoreParameters, it.txt.ens.core.X509CertificateRetrievalParameters, java.io.File, it.txt.ens.schema.request.factory.ENSAuthorisationRequestFactory, it.txt.ens.schema.response.factory.ENSAuthorisationResponseFactory, it.txt.ens.core.factory.ENSConnectionParametersFactory, it.txt.ens.client.core.factory.ENSEventFactory)
         */
        @Override
        public ENSSubscriber create(String subjectID,
                ENSResource targetResource,
                Date sessionExpiration,
                ENSAuthzServiceConnectionParameters accessRequestBrokerConnParams,
                KeystoreParameters keystoreParams,
                X509CertificateRetrievalParameters certParams,
                File capabilityDirectory,
                ENSAuthorisationRequestFactory requestFactory,
                ENSAuthorisationResponseFactory responseFactory,
                ENSBrokerConnectionParametersFactory connParamsFactory,
                ENSEventFactory eventFactory) {
            return new BasicENSSubscriber(subjectID, targetResource, sessionExpiration,
                    accessRequestBrokerConnParams, keystoreParams, certParams, capabilityDirectory, 
                    requestFactory, responseFactory, connParamsFactory, eventFactory);
        }
    }
    
    private class BasicENSConsumer extends DefaultConsumer{
        /**
         * @param channel
         */
        public BasicENSConsumer(Channel channel) {
            super(channel);
        }

        /* (non-Javadoc)
         * @see com.rabbitmq.client.DefaultConsumer#handleDelivery(java.lang.String, com.rabbitmq.client.Envelope, com.rabbitmq.client.AMQP.BasicProperties, byte[])
         */
        @Override
        public void handleDelivery(String consumerTag,
                Envelope envelope, BasicProperties properties,
                byte[] body) throws IOException {
            ENSEvent event = eventFactory.create(envelope.getRoutingKey(), 
                properties.getHeaders(), body, properties.getContentType(), properties.getContentEncoding(),
                properties.getPriority(), properties.getTimestamp(), 
                properties.getDeliveryMode() == AMQPConstants.NOT_PERSISTENT_DELIVERY ? false : true,
                properties.getAppId());
            if (LOGGER.isLoggable(Level.FINER))
                LOGGER.log(Level.FINER, "Received event " + event.toString());
            for (Entry<String, ENSEventListener> listenerEntry : listeners.entrySet())
                listenerEntry.getValue().onEvent(event);
        }

        /* (non-Javadoc)
         * @see com.rabbitmq.client.DefaultConsumer#handleShutdownSignal(java.lang.String, com.rabbitmq.client.ShutdownSignalException)
         */
        @Override
        public void handleShutdownSignal(String consumerTag,
                ShutdownSignalException sig) {
            if (LOGGER.isLoggable(Level.WARNING))
                LOGGER.log(Level.WARNING, "ensShutdownSignal",this.getConsumerTag());
            try {
                //cannot unsubscribe because the channel is not working
                isSubscribed = true;
                disconnect();
            } catch (IllegalStateException e) {
                closeConnection();
            }
        }
    }
}
